package com.example.springbootmqtt.messaging;

import com.example.springbootmqtt.mqtt.MqttClientFactory;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.lifecycle.MqttClientDisconnectedContext;
import com.hivemq.client.mqtt.lifecycle.MqttClientDisconnectedListener;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.function.Consumer;



@Slf4j
@RequiredArgsConstructor
public class MqttDefaultListener implements MqttClientDisconnectedListener {

    private final MqttClientFactory mqttClientFactory;

    private Mqtt5AsyncClient client;

    // in case of connection issues how long we wait till we retry connection to the broker
    private long retryTimeout = 10000;

    /**
     * Start the MQTT client ,subscribe to topic and start handling messages
     *
     * @param initialStart if true the method will fail immediately if it cannot connect to the broker
     */
    public void startProcess(boolean initialStart) {
        boolean isConnected = false;
        while (!isConnected) {
            try {
                log.info("Connecting to MQTT broker on localhost");
                client = mqttClientFactory.createMqtt5Client(this).toAsync();
                // ensure we continue the previous session meaning we will get the messages that were sent to the broker while we were offline
                client.connectWith().cleanStart(false).noSessionExpiry().send().get();

                log.info("Subscribing to topic testtopic2");
                client.subscribeWith()
//                        .topicFilter("testtopic2/#")
                        .topicFilter("$share/XID_MDK_FULFILL/testtopic2/#")
                        .qos(MqttQos.AT_LEAST_ONCE)
                        .callback(callbackHandler())
                        .send().get();

                isConnected = true;
            } catch (InterruptedException e) {
                log.error("Mqtt listening client interrupted", e);
                throw new RuntimeException(e);
            } catch (Exception e) {
                // if the server starts up fail immediately if broker is not accessible
                if (initialStart) {
                    log.error("Could not connect to mqtt client", e);
                    throw new RuntimeException(e);
                }

                log.error("Could not connect to mqtt client,retrying in: {}ms", retryTimeout, e);
                try {
                    //noinspection BusyWait
                    Thread.sleep(retryTimeout);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    private Consumer<Mqtt5Publish> callbackHandler() {
        return mqtt5Publish -> {
            // publish 类型是 Mqtt5Publish
            String topic = mqtt5Publish.getTopic().toString();
            String payload = new String(mqtt5Publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
            System.out.println("📩 收到消息：topic = " + topic + "，payload = " + payload);
        };
    }


    @PreDestroy
    public void destroy() {
        if (client != null) {
            client.disconnect();
        }
    }

    @Override
    public void onDisconnected(MqttClientDisconnectedContext context) {
        log.warn("Disconnected from MQTT broker on: {} . Source: {}, Cause: {}", context.getClientConfig().getServerAddress(), context.getSource(), context.getCause().toString());
        startProcess(false);
    }
}
